package com.example.mq.server.consumermanager;

import com.example.mq.common.MqException;
import com.example.mq.server.messagemanager.Message;
import com.example.mq.server.virtualhost.VirtualHost;
import com.example.mq.server.virtualhost.MQueue;

import java.io.IOException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class ConsumerManager {

    private VirtualHost virtualHost;

    // 存放令牌的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();

    // 扫描线程
    private Thread scannerThread = null;

    private ExecutorService workerPool = Executors.newFixedThreadPool(4);

    public ConsumerManager (VirtualHost virtualHost) {
        this.virtualHost = virtualHost;

        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 拿到令牌
                    String queueName = tokenQueue.take();
                    // 2. 根据令牌, 找到队列
                    MQueue queue = virtualHost.getMemoryData().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取令牌后发现, 该队列名不存在! queueName=" + queueName);
                    }
                    // 3. 从这个队列中消费一个消息.
                    synchronized (queue) {
                        preConsume(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        // 把线程设为后台线程.
        scannerThread.setDaemon(true);
        scannerThread.start();
    }
    // 当生产消息后，这个方法就会被调用，从而触发消费消息的机制
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    // 给指定的一个队列新增一个订阅的消费者
    public void addConsumer(String consumerTag, String queueName) throws MqException, IOException, ClassNotFoundException {
        // 找到对应的队列.
        MQueue queue = virtualHost.getMemoryData().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有了一些消息了, 需要立即就消费掉.
            int n = virtualHost.getMemoryData().getMessageCount(queueName);

            System.out.println(consumerTag + " 想订阅的队列中有 " + n + " 条消息");

            for (int i = 0; i < n; i++) {
                // 这个方法调用一次就消费一条消息.
                virtualHost.consumeMessage(consumerTag, virtualHost.getMemoryData().pollMessage(queueName), queue);
            }
        }
    }

    // 包装消费工作
    public void preConsume(MQueue queue) {
        // 1. 轮询，拿到消费者
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            // 当前队列没有消费者, 暂时不消费. 等后面有消费者出现再说.
            return;
        }
        // 2. 从队列中取出一个消息
        Message message = virtualHost.getMemoryData().pollMessage(queue.getName());
        if (message == null) {
            // 当前队列中还没有消息, 也不需要消费.
            return;
        }
        // 3. 消费消息
        workerPool.submit(() -> {
            try {

                virtualHost.consumeMessage(luckyDog.getConsumerTag(), message, queue);

                // 删除内存中的消息
                virtualHost.getMemoryData().removeMessage(message.getMessageId());
                System.out.println("[ConsumerManager] 消息被成功消费! queueName=" + queue.getName());
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}
